Skip to content

subscription_thread_dispatcher: better callback/thread lifecycle - #195

Open
stephen-derosa wants to merge 8 commits into
livekit:mainfrom
stephen-derosa:fix/subscription_thread_dispatcher_cleanup
Open

stephen-derosa wants to merge 8 commits into
livekit:mainfrom
stephen-derosa:fix/subscription_thread_dispatcher_cleanup

Conversation

@stephen-derosa

@stephen-derosa stephen-derosa commented Jul 6, 2026

Copy link
Copy Markdown
Collaborator

Overview

Fixes #235 and hardens reader-thread lifecycle in SubscriptionThreadDispatcher. Frame callbacks now start a reader regardless of when they are registered, replacement actually replaces, and driving the dispatcher from inside a frame callback no longer self-joins. No public API changessetOn*FrameCallback now does what it always documented.

The bugs

  • Late registration never started a reader (Calling setOnVideo/AudioFrameCallback outside of the onTrackSubscribed handler never starts a frame thread handler #235). handleTrackSubscribed ran once, right after RoomDelegate::onTrackSubscribed, and no-oped without a registered callback. Registering afterwards (e.g. from a GUI thread) stored the callback and nothing else happened.
  • Replacement was a silent no-op. Reader threads capture the callback by value. Re-registering only overwrote the dispatcher's copy; the running thread kept invoking the old callback.
  • Self-joins terminated the process. Room::disconnect() or removeOnDataFrameCallback called from inside a data frame callback joined the calling thread: std::system_error, then a still-joinable std::thread destroyed during unwinding.

Changes

  • Retained subscriptions. The dispatcher keeps the subscribed track per (participant, track_name) until unsubscribed, so setOn* starts a reader immediately when the track is already subscribed and defers otherwise. Room also resolves the current publication, which covers registration from inside onTrackSubscribed.
  • Replace in place, join before start. Every stop path (setOn*, clearOn*, resubscribe with a new SID, unsubscribe) extracts the reader under the lock, marks the key draining, joins outside the lock, then restarts. No caller can start a reader for a draining key, so old and new callbacks never run concurrently and concurrent setters cannot pile up untracked threads.
  • Duplicate subscription events. A repeated track_subscribed for the SID a reader already serves is a no-op; a new SID (republish) stops the old reader first.
  • Detach on self-join. No reader lambda captures this; data readers report exit through their own shared state. All readers are therefore safe to detach, so a lifecycle call from inside the reader's own callback logs a warning, detaches, and still takes effect.
  • Data track teardown. In-flight RemoteDataTrack::subscribe() calls are marked cancelled so unpublish/republish cannot leave a stale reader or duplicate FFI subscription behind.
  • SubscriptionThreadDispatcher deprecated for direct use (LIVEKIT_DEPRECATED); it stays exported to avoid a major bump.

Behavioral notes

  • setOn*/clearOn* block until any in-flight invocation of the previous callback returns. A callback that never returns blocks registration indefinitely.
  • Re-entrant calls are discouraged but handled. From inside a frame callback for the same key the SDK warns and detaches that reader. The in-flight invocation of the previous callback finishes only after the call returns, so the no-overlap guarantee does not hold for that one invocation.

Public API

No signature changes. include/livekit/room.h changes are documentation only. Callback type aliases moved to include/livekit/frame_callbacks.h (still reachable through room.h).

Test-only: RemoteDataTrack::testFfiHandleId() removed in favor of RemoteDataTrackTestAccess; RoomTestAccess exposes reader/drain counts and retained-subscription state; RemoteDataTrack's private constructor is LIVEKIT_INTERNAL_API so tests can link it.

Testing

  • Unit (SubscriptionThreadDispatcherTest, 91): Calling setOnVideo/AudioFrameCallback outside of the onTrackSubscribed handler never starts a frame thread handler #235 in both orderings for audio/video/video-event; replacement semantics including the shared video slot; SID dedup and republish; drain marks set and cleared by every lifecycle path, with starts deferred while draining; self-join handling for setOn*, clearOn*, removeOnDataFrameCallback, and stopAll from both media and data reader threads; data reader cancellation and finished-state handling.
  • Integration (FrameCallbackServerTest, 5): the exact Calling setOnVideo/AudioFrameCallback outside of the onTrackSubscribed handler never starts a frame thread handler #235 scenario for video, video-event, and audio; clear-and-re-register without a new event; registration from inside onTrackSubscribed starting exactly one reader.
  • Integration (FrameCallbackReplacementTest, 14): delivery switches on replacement; setter blocks for a slow callback without overlapping it; sequential and concurrent churn leave exactly one reader; deferred start and republish; re-entrant setOn*/clearOn*/data removal take effect; Room::disconnect() from inside a data callback.
  • Full unit suite (412) and the data-track, room-event, and platform-audio integration suites (37) pass locally against a LiveKit server. clang-format and clang-tidy are clean on the changed files.

@stephen-derosa stephen-derosa self-assigned this Jul 6, 2026
Copilot AI review requested due to automatic review settings July 6, 2026 21:39
@stephen-derosa
stephen-derosa marked this pull request as draft July 6, 2026 21:39

Copilot AI left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

This PR refines SubscriptionThreadDispatcher’s reader-thread lifecycle handling to avoid redundant restarts and to ensure data reader threads reliably tear down (especially when a data-track subscription is still in flight), aligning with the SDK’s threading model by preventing teardown hangs.

Changes:

  • Add cancellation signaling for active data readers and ensure cancellation is set before closing/joining during unpublish and teardown.
  • Introduce a self-cleanup path (eraseDataReaderIfCurrent) so data reader threads can safely remove stale slots when they exit.
  • Skip redundant audio/video/data reader restarts when the same track SID is already active.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated no comments.

File Description
src/tests/unit/test_subscription_thread_dispatcher.cpp Adds unit tests covering data reader cancellation semantics and self-erase behavior.
src/subscription_thread_dispatcher.cpp Implements cancellation-before-close, self-erase cleanup, and redundant reader-start suppression.
include/livekit/subscription_thread_dispatcher.h Adds per-reader state (track_sid, cancelled) and declares the new cleanup helper.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

devin-ai-integration[bot]

This comment was marked as resolved.

@stephen-derosa stephen-derosa changed the title rm unneeded function, better cleanup subscription_thread_dispatcher: guard against duplicate room events Jul 7, 2026
@stephen-derosa
stephen-derosa force-pushed the fix/subscription_thread_dispatcher_cleanup branch from 2f313fe to f1ff6e6 Compare July 7, 2026 18:17
@stephen-derosa
stephen-derosa requested a review from Copilot July 7, 2026 22:11
@stephen-derosa
stephen-derosa marked this pull request as ready for review July 7, 2026 22:11

This comment was marked as resolved.

devin-ai-integration[bot]

This comment was marked as resolved.

This comment was marked as resolved.

devin-ai-integration[bot]

This comment was marked as resolved.

This comment was marked as resolved.

devin-ai-integration[bot]

This comment was marked as resolved.

This comment was marked as resolved.

Comment thread include/livekit/remote_data_track.h Outdated
Comment thread include/livekit/room.h Outdated
/// @return @c true if the callback was registered; @c false if a reader is
/// already active for the key (call @ref clearOnAudioFrameCallback
/// first) or the room has no dispatcher.
[[nodiscard]] bool trySetOnAudioFrameCallback(const std::string& participant_identity, const std::string& track_name,

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

From what I can tell, these new "try" methods have the same inputs as the deprecated prior ones, just a new return type (and name).

This is actually a situation where I think maintaining the API and using exceptions is cleaner than deprecation/new API just to get the bool return type.

Thoughts on this? Could we not move the new implementations from room.cpp into the older versions and throw instead of returning false?

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

isnt that still changing the public API though? maybe im wrong but i would think adding a throw to a function that previously didnt would require a major bump?

@alan-george-lk alan-george-lk Jul 8, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fair point, technically speaking those methods were never marked noexcept and anything within them could throw (STL container exceptions, runtime exceptions from FFI stuff, etc.) so I think it's the least of the evils?

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

hm yeah im a little torn here. I dont want to just start throwing underneath applications, especially when our mantra has somewhat been team #nothrow 😅

but the point of other things can throw inside the function are valid 🤔

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also torn, I don't love my suggestion, to me keeping the API clean/avoiding deprecations is the least of the evils. Maybe @xianshijing-lk can chime in

Comment thread include/livekit/room.h Outdated
/// @return @c true if the callback was registered; @c false if a reader is
/// already active for the key (call @ref clearOnVideoFrameCallback
/// first) or the room has no dispatcher.
[[nodiscard]] bool trySetOnVideoFrameCallback(const std::string& participant_identity, const std::string& track_name,

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess I didn't understand why these callback events need to be in the room ?
couldn't they be in the remote participants?

And does this frame callbacks work on both local / remote participants ?

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the room has access to the room events which are needed to set callbacks to incoming track messages -- the remote participant doesnt really.

This callback only works for local participants, the intention is to provide a nice ergonomic way for the user to set callbacks for incoming messages.

Comment thread src/room.cpp
} else {
// The track is not subscribed yet. The callback is registered; the reader
// starts when the track is subscribed (see kTrackSubscribed in onEvent).
LK_LOG_DEBUG(

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should you still return true here ?

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

good question, yes since the user has successfully set the callback. It doesnt necessarily mean that a track has been published or that they will get messages on said track. If we were to only return true to a callback being set after a track has been published, then early joining participants that want incoming messages will effectively have to periodically set callbacks until success, or listen to room events -- both of which defeat the purpose of the ergonomic functionality of these setOn*Callback functions.

Comment thread src/room.cpp
// If we've already subscribed to the track, handle it immediately
auto track = findSubscribedRemoteTrack(participant_identity, track_name);
if (track) {
subscription_thread_dispatcher_->handleTrackSubscribed(participant_identity, track_name, track);

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think I forgot some technical details here, is this subscription_thread_dispatcher_ a manager that manages all the subscription thread ?
and does one callback has one corresponding subscription thread ? or one thread that handles all the callbacks ?

@stephen-derosa stephen-derosa Jul 9, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think I forgot some technical details here, is this subscription_thread_dispatcher_ a manager that manages all the subscription thread ?

it manages subscribing to tracks and providing incoming frames to the users callbacks

and does one callback has one corresponding subscription thread ? or one thread that handles all the callbacks ?

Each subscription gets its own thread

devin-ai-integration[bot]

This comment was marked as resolved.

@stephen-derosa
stephen-derosa force-pushed the fix/subscription_thread_dispatcher_cleanup branch from 2179d4c to fbdcbd4 Compare July 9, 2026 16:21
Comment thread include/livekit/subscription_thread_dispatcher.h Outdated
@stephen-derosa

Copy link
Copy Markdown
Collaborator Author

ensure that this also works for local publishing of video/audio

@stephen-derosa
stephen-derosa force-pushed the fix/subscription_thread_dispatcher_cleanup branch from 4f18b6f to 2aa6d4c Compare July 17, 2026 19:56
@stephen-derosa
stephen-derosa requested a review from a team as a code owner August 28, 2026 20:46
@stephen-derosa
stephen-derosa force-pushed the fix/subscription_thread_dispatcher_cleanup branch from 4b75ba8 to 0dbdc64 Compare August 28, 2026 20:48
devin-ai-integration[bot]

This comment was marked as resolved.

Comment thread include/livekit/subscription_thread_dispatcher.h Outdated
@stephen-derosa
stephen-derosa force-pushed the fix/subscription_thread_dispatcher_cleanup branch from ab8df3a to bb9224e Compare September 8, 2026 23:18
…plicate room events test

SubscriptionThreadDispatcher: proper replacing of audio/video callbacks. Deprecate a setOn*Callback(), replace with trySetOn*Callback()

fix thread detaching
@stephen-derosa
stephen-derosa force-pushed the fix/subscription_thread_dispatcher_cleanup branch from 27e9cae to f5d5055 Compare September 14, 2026 17:15
devin-ai-integration[bot]

This comment was marked as resolved.

@devin-ai-integration devin-ai-integration Bot left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Devin Review found 3 new potential issues.

7 flags not posted on this PR by your GitHub settings — view them in Devin Review. (Configure)

Devin Review

Comment on lines +500 to +506
/// Currently subscribed remote audio/video tracks keyed by @ref CallbackKey.
std::unordered_map<CallbackKey, std::shared_ptr<Track>, CallbackKeyHash> subscribed_tracks_;

/// Keys whose previous reader has been extracted but not yet joined, with the
/// number of such in-progress drains. No reader is started for a key while
/// it has an entry here. See @ref extractReaderForDrainLocked.
std::unordered_map<CallbackKey, int, CallbackKeyHash> draining_readers_;

@devin-ai-integration devin-ai-integration Bot Sep 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔴 Dispatcher ABI corrupts existing clients

Existing clients allocate the old SubscriptionThreadDispatcher size, but its constructor now initializes a larger layout. The added maps write beyond caller-allocated storage, causing memory corruption or crashes.

Learn more

SubscriptionThreadDispatcher is a public, exported class with an out-of-line constructor. C++ clients allocate its storage using the size recorded when they compiled. Adding data members changes that size without changing the constructor symbol, so an old executable can load the new shared library and call the new constructor on an undersized object. The repository's public ABI rule requires backwards compatibility.

Example: An application compiled against the previous release embeds SubscriptionThreadDispatcher dispatcher;. After replacing only liblivekit.so, the new constructor initializes subscribed_tracks_ and draining_readers_ beyond the application's allocated object.

Recommended fix: Preserve the released class layout. Move evolving dispatcher state behind an ABI-stable private implementation pointer, or defer this layout change to a declared ABI-breaking major release.

Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

Comment on lines +130 to +140
old_thread = extractReaderForDrainLocked(key);
const bool replacing = audio_callbacks_.find(key) != audio_callbacks_.end();
audio_callbacks_[key] = RegisteredAudioCallback{std::move(callback), opts};
LK_LOG_DEBUG(
"Registered audio frame callback for participant={} track_name={} "
"replacing_existing={} stopped_reader={} total_audio_callbacks={}",
participant_identity, track_name, replacing, old_thread.joinable(), audio_callbacks_.size());
}
// Joins the previous reader first, then starts a fresh one bound to the new
// callback if the track is subscribed.
finishReaderDrainAndRestart(key, std::move(old_thread), "setOnAudioFrameCallback");

@devin-ai-integration devin-ai-integration Bot Sep 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Concurrent disconnect leaves callback running

Concurrent disconnect() misses a reader extracted by setOn*FrameCallback while that setter waits for its callback. Disconnect returns before the callback ends, so teardown does not quiesce frame delivery.

Learn more

Replacement removes the active reader from active_readers_ before joining it. The setter then owns the only std::thread handle while the callback can remain in flight. A concurrent stopAll scans only active_readers_, so it cannot close or join this draining reader and can return first. The dispatcher is documented as thread-safe, and stopAll() is the shutdown path that must stop all readers.

Example: Callback A blocks for ten seconds. Thread 1 replaces A and waits in join(). Thread 2 calls Room::disconnect(); stopAll() sees no active reader and returns while A keeps running.

Recommended fix: Make shutdown coordinate with in-progress drains. Track drain completion with shared reader state or a condition variable, and make stopAll() wait for every non-self draining reader without holding lock_.

Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

Comment on lines +264 to +265
subscribed_tracks_.erase(key);
old_thread = extractReaderForDrainLocked(key);

@devin-ai-integration devin-ai-integration Bot Sep 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Stale unsubscribe stops new publication

A delayed unsubscribe makes handleTrackUnsubscribed erase the current same-name subscription without comparing SIDs. The new publication's reader stops and receives no restart event.

Learn more

The dispatcher retains media subscriptions by participant identity and track name, but a subscription event also identifies a concrete publication by SID. Room event forwarding resolves the unsubscribe SID, then discards it and passes only the publication name. If an old unsubscribe is delivered after a same-name republish or resubscribe, these lines erase and stop whichever SID is currently retained.

Example: Alice's camera publication TR_old is replaced by TR_new. After TR_new starts its reader, a delayed unsubscribe for TR_old reaches the dispatcher and removes TR_new, leaving the callback silent.

Recommended fix: Pass the unsubscribed SID into handleTrackUnsubscribed. Under lock_, erase and drain only when the retained track and active reader still match that SID.

Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

@stephen-derosa stephen-derosa changed the title subscription_thread_dispatcher: guard against duplicate room events subscription_thread_dispatcher: better callback/thread lifecycle Sep 14, 2026
std::unordered_map<CallbackKey, ActiveReader, CallbackKeyHash> active_readers_;

/// Currently subscribed remote audio/video tracks keyed by @ref CallbackKey.
std::unordered_map<CallbackKey, std::shared_ptr<Track>, CallbackKeyHash> subscribed_tracks_;

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

will this create races or confusion to our users regarding the subscribed tracks ?

I think our room has API about RemoteTrackPublication, that tracks whether a publication is subscribed. And this |subscribed_tracks_| might confuse developers on which one is the ground truth regarding the subscribed tracks ?

or are they strictly in sync?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Calling setOnVideo/AudioFrameCallback outside of the onTrackSubscribed handler never starts a frame thread handler

4 participants